package com.smartcommunity.kafka;

import com.alibaba.fastjson.JSON;
import com.smartcommunity.base.common.CacheMap;
import com.smartcommunity.entity.statistic.StatisticsMsgData;
import com.smartcommunity.entity.statistic.StatisticsMsg;
import com.smartcommunity.utils.DateUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.util.List;

@Component
@Slf4j
public class StatisticsConsumer {

    @KafkaListener(topics = {"${kafka.business.topic.statistics:statistics}"},
            groupId = "${kafka.statistics.consumer.group:group_statistics1}",
            containerFactory = "statisticsListenerContainerFactory")
    public void listen(List<String> messages) {
        try {
            messages.forEach(msg -> {
                StatisticsMsg statisticsMsg = JSON.parseObject(msg, StatisticsMsg.class);
                if(null == statisticsMsg || null == statisticsMsg.getData()){
                    log.error("统计数据转换失败,JSON:{}", msg);
                }
                StringBuilder statisticsKeyBuilder = new StringBuilder();
                StatisticsMsgData statisticsMsgData = statisticsMsg.getData();
                statisticsKeyBuilder.append(DateUtil.getDate(statisticsMsgData.getToDayTime(),DateUtil.DATE_HOUR_FORMAT)).append("|");
                statisticsKeyBuilder.append(statisticsMsg.getType()).append("|");
                statisticsKeyBuilder.append(statisticsMsgData.getCountType());
                if(CacheMap.STATISTICS_MAP.containsKey(statisticsKeyBuilder.toString())){
                    Long count = CacheMap.STATISTICS_MAP.get(statisticsKeyBuilder.toString()) + statisticsMsgData.getIncrCount();
                    CacheMap.STATISTICS_MAP.put(statisticsKeyBuilder.toString(),count);
                }else{
                    CacheMap.STATISTICS_MAP.put(statisticsKeyBuilder.toString(), statisticsMsgData.getIncrCount());
                }
            });
        } catch (Exception e) {
            log.error("统计数据转换异常,error:", e);
        }
    }
}